package com.asiainfo.dacp.datastash.broker.facade.service;

import com.asiainfo.dacp.datastash.broker.core.BrokerContext;
import com.asiainfo.dacp.datastash.broker.core.BrokerException;
import com.asiainfo.dacp.datastash.broker.core.StashUnitModel;
import com.asiainfo.dacp.datastash.broker.core.enums.UnitStateEnum;
import com.asiainfo.dacp.datastash.broker.core.vo.StashUnit;
import com.asiainfo.dacp.datastash.broker.core.vo.StashUnitDefine;
import com.asiainfo.dacp.datastash.broker.facade.state.UnitStateLogService;
import com.asiainfo.dacp.datastash.broker.tracelog.StashTraceLogVo;
import com.asiainfo.dacp.datastash.broker.tracelog.StashTraceLogger;
import com.asiainfo.dacp.datastash.orm.po.state.DataStashUnitStateLogPo;

import org.apache.commons.lang3.tuple.Pair;
import com.google.common.base.Strings;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.*;

/**
 * /** 计算需要处理的任务单元，需要根据两方的状态同步redis与db日志
 *
 * 增加的单元需要新增到任务待处理队列（表）中
 *
 * @author MeiKefu
 * @date 2017-12-13
 */
@Service
public class UnitLoaderService {

	@Autowired
	private StashTraceLogger tracelog;
	private String clzName = this.getClass().getSimpleName();

	@Autowired
	private StashUnitModel stashUnitModel;

	@Autowired
	private UnitStateLogService unitStateService;

	@Autowired
	private UnitInitService unitInitService;

	/**
	 * 创建一个StashUnit任务
	 *
	 * 需要对比 redis与数据库中的状态是否一致
	 */
	public void createStashUnit(String unitCode, String batchNo, boolean isTurnover) {
		
		String unitProcId = null;
		StashUnit stashUnit = buildStashUnit(unitCode, batchNo, isTurnover, unitProcId);
		unitStateService.create(stashUnit);
		
		/*
		// 1.先查找数据库中是否有这个状态
		Pair<String, String> unitProcIdAndState = unitStateService.queryUnitState(unitCode, batchNo);

		if (unitProcIdAndState != null) {
			// 数据库存在，判断redis是否存在，不存在从数据库中恢复
			String unitProcId = unitProcIdAndState.getLeft();
			String state = unitProcIdAndState.getRight();
			tracelog.debug(StashTraceLogVo.buildStashTraceLogVo(clzName, "createStashUnit", "数据库中存在数据，需要进行校验", unitCode,
					batchNo, unitProcId));

			StashUnit memUnitState = stashUnitModel.getStashUnit(unitProcId);

			if (memUnitState != null) {
				if (!memUnitState.getState().equals(state)) {
					// redis状态与数据库不一致，已redis为准,需要更新数据库的状态
					if (UnitStateEnum.Failed.value().equals(memUnitState.getState())
							&& UnitStateEnum.Initialization.value().equals(state)) {
						tracelog.warn(StashTraceLogVo.buildStashTraceLogVo(memUnitState, clzName, "createStashUnit",
								"数据库与内存状态不一致，判断为手工重做，使用数据库中的状态变更内存状态为初始"));
						// 如果内存为错误状态，数据库状态为待运行，证明是手工重做任务，已数据库为准
						memUnitState.initial();
					} else {
						try {
							tracelog.warn(StashTraceLogVo.buildStashTraceLogVo(memUnitState, clzName, "createStashUnit",
									"数据库与内存状态不一致，已内存状态为准变更数据库状态"));
							unitStateService.update(memUnitState.getUnitProcId(), memUnitState.getState(),
									"数据库状态与内存不一致");
						} catch (BrokerException e) {
							e.printStackTrace();
						}
					}
				} else {
					tracelog.debug(StashTraceLogVo.buildStashTraceLogVo(memUnitState, clzName, "createStashUnit",
							"数据库中与内存状态一致"));
				}
			} else {
				tracelog.warn(StashTraceLogVo.buildStashTraceLogVo(clzName, "createStashUnit",
						"数据库中存在数据，内存丢失数据，重新恢复内存中数据", unitCode, batchNo, unitProcId));
				buildStashUnit(unitCode, batchNo, isTurnover, unitProcId);
			}
		} else {
			// 判断内存中有没有谁
			String unitProcId = null;
			
			
			List<StashUnit> stashUnits = stashUnitModel.getStashUnitList();
			// TODO 可能影响性能
			if (stashUnits != null && stashUnits.size() > 0) {
				for (StashUnit stashUnit : stashUnits) {
					if (stashUnit.getUnitCode().equals(unitCode) && stashUnit.getBatchNo().equals(batchNo)) {
						unitProcId = stashUnit.getUnitProcId();
						break;
					}
				}
			}
			
			
			StashUnit stashUnit = buildStashUnit(unitCode, batchNo, isTurnover, unitProcId);
			
			
			if (unitProcId == null) {
				tracelog.info(StashTraceLogVo.buildStashTraceLogVo(stashUnit, clzName, "createStashUnit", "初始化任务成功"));
			} else {
				tracelog.warn(
						StashTraceLogVo.buildStashTraceLogVo(stashUnit, clzName, "createStashUnit", "内存中包含，数据库不存在"));
			}
			
			
			unitStateService.create(stashUnit);
		}*/
		
		
	}

	private StashUnit buildStashUnit(String unitCode, String batchNo, boolean isTurnover, String unitProcId) {
		StashUnit stashUnit = new StashUnit(unitCode, batchNo, isTurnover, unitProcId);
		stashUnitModel.putUnitIfAbsent(stashUnit);
		return stashUnit;
	}

	/**
	 * 找到所有需要处理的任务批次，可以有多个批次
	 */
	public void unitLoad() {

		Map<String, StashUnitDefine> memUnits = BrokerContext.getInstance().getStashUnitDefs();

		if (memUnits == null || memUnits.size() == 0) {
			tracelog.error(
					new StashTraceLogVo(clzName, "findUnitBatch:error", " BrokerContext.getStashUnitDefines没有值"));
			return;
		}

		for (StashUnitDefine memUnit : memUnits.values()) {

			// TODO
			// 计算所有需要做的任务的批次号，放入状态队列，如果不排她同一单元多个批次都可以同时，解析当前批次以前的所有任务批次都放入队列

			String redisCurBatchNo = stashUnitModel.getCurrentUnitBatchNo(memUnit.getUnitCode());

			if (Strings.isNullOrEmpty(redisCurBatchNo)) {
				tracelog.warn(new StashTraceLogVo(clzName, "findUnitDefsBatchNo:fail", "获取当前单元的批次失败",
						memUnit.getUnitCode(), "", ""));
				// continue;
				// fix:: 如果不存在，则更新数据库最新到内存
				unitInitService.updateUnitBatchNo(memUnit);
			}

			// 应该执行的批次
			String shouldBatchNo = lastBatchNo(GregorianCalendar.getInstance(), memUnit.getBatchType(),
					memUnit.getBatchOffset());
			if (shouldBatchNo == null) {
				// 不支持的批次执行
				continue;
			}

			long nShouldBatchNo = Long.valueOf(shouldBatchNo);
			long nRedisCurBatchNo = Long.valueOf(redisCurBatchNo);

			if (nRedisCurBatchNo <= nShouldBatchNo) {

				// 找当前批次是否在执行
				if (batchNotProcessing(memUnit.getUnitCode(), redisCurBatchNo)) {
					createStashUnit(memUnit.getUnitCode(), redisCurBatchNo, true);
				}

				/* 
				 * 并发流程，不允许翻牌， 在接口服务中处理
				if (!memUnit.isSequential()) {
					List<String> prepareBatchNos = calAllBatchNo(nRedisCurBatchNo, memUnit.getBatchStep(),
							memUnit.getBatchType());// 并发执行是倒序的
					// 并发执行的任务，不翻牌子
					// TODO 批次的排序问题
					for (String prepareBatchNo : prepareBatchNos) {
						createStashUnit(memUnit.getUnitCode(), prepareBatchNo, false);
					}
				}

				*/
			}

		}
	}

	public static String lastBatchNo(String curBatchNo, String batchType, int batchOffset) throws ParseException {
		Calendar calendar = Calendar.getInstance();
		SimpleDateFormat sdf = null;

		if ("daily".equals(batchType)) {
			sdf = new SimpleDateFormat("yyyyMMdd");

		} else if ("monthly".equals(batchType)) {
			sdf = new SimpleDateFormat("yyyyMM");
		} else if ("hourly".equals(batchType)) {
			sdf = new SimpleDateFormat("yyyyMMddHH");
		} else if ("minutely".equals(batchType)) {
			sdf = new SimpleDateFormat("yyyyMMddHHmm");
		}
		calendar.setTime(sdf.parse(curBatchNo));
		return lastBatchNo(calendar, batchType, batchOffset);
	}

	public static String lastBatchNo(Calendar calendar, String batchType, int batchOffset) {

		SimpleDateFormat sdf = null;

		if ("daily".equals(batchType)) {
			sdf = new SimpleDateFormat("yyyyMMdd");
			calendar.add(Calendar.DATE, batchOffset);
		} else if ("monthly".equals(batchType)) {
			sdf = new SimpleDateFormat("yyyyMM");
			calendar.add(Calendar.MONTH, batchOffset);
		} else if ("hourly".equals(batchType)) {
			sdf = new SimpleDateFormat("yyyyMMddHH");
			calendar.add(Calendar.HOUR, batchOffset);
		} else if ("minutely".equals(batchType)) {
			sdf = new SimpleDateFormat("yyyyMMddHHmm");
			calendar.add(Calendar.MINUTE, batchOffset);
		}
		return sdf != null ? sdf.format(calendar.getTime()) : null;
	}

	/**
	 * 判断该批次是否执行
	 */
	private boolean batchNotProcessing(String unitCode, String batchNo) {
		
		List<DataStashUnitStateLogPo> queryNonOverUnitState = unitStateService.queryNonOverUnitState(unitCode, batchNo);
		
		if(queryNonOverUnitState == null || queryNonOverUnitState.size() == 0){
			return true;
		}else{
			return false;
		}
		
		/*
		List<StashUnit> redisUnitStates = stashUnitModel.getStashUnitList();

		if (redisUnitStates != null && redisUnitStates.size() > 0) {
			for (StashUnit redisUnitState : redisUnitStates) {
				if (redisUnitState.getUnitCode().equals(unitCode) && redisUnitState.getBatchNo().equals(batchNo)) {
					return true;
				}
			}
		}

		return false;
		*/
	}

	/**
	 * 计算差值的批次
	 */
	private List<String> calAllBatchNo(int nRedisCurBatchNo, int batchStep, String batchType) {
		List<String> result = new ArrayList();
		Calendar calendar = GregorianCalendar.getInstance();
		String _batchNo = lastBatchNo(calendar, batchType, 0);
		int nLastBatchNo = Integer.valueOf(_batchNo);
		do {
			result.add(String.valueOf(nLastBatchNo));
			nLastBatchNo = Integer.valueOf(lastBatchNo(calendar, batchType, -batchStep));
		} while (nLastBatchNo > nRedisCurBatchNo);

		return result;
	}

}
